package cn._51doit.flink.day12;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * 使用FlinkSQL通过广告曝光、点击的人数和次数
 */
public class SQLAdvCount {

    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        env.enableCheckpointing(50000);


        //使用KafkaConnector从Kafka中读取数据
        tEnv.executeSql(
                "CREATE TABLE tb_user_order (\n" +
                        "  `uid` STRING,  \n" +
                        "  `aid` STRING,  \n" +
                        "  `type` STRING  \n" +
                        ") WITH (\n" +
                        "  'connector' = 'kafka',\n" +
                        "  'topic' = 'adv-log',\n" +
                        "  'properties.bootstrap.servers' = 'node-1.51doit.cn:9092,node-2.51doit.cn:9092,node-3.51doit.cn:9092',\n" +
                        "  'properties.group.id' = 'testGroup',\n" +
                        "  'scan.startup.mode' = 'earliest-offset',\n" +
                        "  'format' = 'csv', \n" +
                        "  'csv.ignore-parse-errors' = 'true'" +
                        ")"
        );

        tEnv.executeSql(
                "CREATE TABLE tb_mysql_sink (\n" +
                        "  `aid`  STRING,  \n" +
                        "  `type` STRING,  \n" +
                        "  `counts1` BIGINT,  \n" +
                        "  `counts2` BIGINT, \n" +
                        "  PRIMARY KEY (aid, type) NOT ENFORCED \n" +
                        ") WITH (\n" +
                        "   'connector' = 'jdbc',\n" +
                        "   'url' = 'jdbc:mysql://node-1.51doit.cn:3306/doit28',\n" +
                        "   'table-name' = 'tb_adv_count',\n" +
                        "   'username' = 'root',\n" +
                        "   'password' = '123456'\n" +
                        ")"
        );

        tEnv.executeSql("INSERT INTO tb_mysql_sink " +
                "SELECT aid, type, count(*) counts1, count(distinct uid) counts2 FROM tb_user_order GROUP BY aid, type");
    }
}
